package com.alipay.traditionalbus;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 20 thread we can get
 * ops/sec    = 56,981,360
 latency ns = 17.550

 * if consumer>1 ,then queue has conflict
 * ops/sec    = 50,460,856
    latency ns = 19.817
 *
 *
 * Created by crazytime on 14-3-1.
 */
public class TraditionalTest {
    private static EventBus bus=new EventBus();


    public static void main(String[] args) throws InterruptedException {

        int publiserNum=20;
        int consumerNum=2;
        long ITERATIONS = 1000L * 1000L * 100L;
        CountDownLatch pub=new CountDownLatch(publiserNum);
        CountDownLatch con=new CountDownLatch(consumerNum);
        AtomicBoolean stop=new AtomicBoolean(false);

        Thread[] pubThread=initPublishThread(publiserNum,ITERATIONS,pub);
        Thread[] conThread=initConsumeThread(consumerNum,stop,con);
        long start=System.nanoTime();
        for(Thread it:conThread){
            it.start();
        }
        for(Thread it:pubThread){
            it.start();
        }
        pub.await();
        //
        stop.set(true);
        con.await();
        long duration=System.nanoTime()-start;

        final long ops = (ITERATIONS * 1000L * 1000L * 1000L) / duration;
        System.out.format("ops/sec    = %,d\n", ops);
        System.out.format("latency ns = %.3f%n", duration / (float)(ITERATIONS) );


    }

    public static Thread[] initPublishThread(int num,long allCounts,CountDownLatch pub){
        Thread[] thread=new Thread[num];
        long step=allCounts/num;
        for(int i=0;i<num;i++){
            thread[i]=new Thread(new EventPublisher(step,1000,bus,pub));
        }
        return thread;
    }

    public static Thread[] initConsumeThread(int num,AtomicBoolean stop,CountDownLatch con){
        Thread[] thread=new Thread[num];
        for(int i=0;i<num;i++){
            thread[i]=new Thread(new EventConsumer(bus,stop,con));
        }
        return thread;
    }


}
